package cn.itcast.flink.join;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * Author itcast
 * Date 2022/1/13 17:22
 * Desc 自定义生成一个 Tuple2<String,1~99>
 * 根据 key 进行分组
 * window操作 ： 5s 划分窗口
 * sum 聚合
 */
public class TimeWindowDemo {
    public static void main(String[] args) throws Exception {
        //todo 执行流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //todo 设置并行度
        env.setParallelism(1);
        //todo 添加数据源
        DataStreamSource<Tuple2<String, Integer>> source = env.addSource(new GenerateRandomNumEverySecond());
        //todo 窗口划分和聚合
        /*SingleOutputStreamOperator<Tuple2<String, Integer>> result = source
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum(1);*/
        //todo 分组、窗口划分和聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> result1 = source.keyBy(t -> t.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum(1);
        //todo 打印结果
        result1.printToErr();
        //todo 执行流环境
        env.execute();
    }

    /*
    自定义Source
    每隔1秒产生一个的k,v  k是hadoop spark flink 其中某一个, v是随机数字
     */
    public static class GenerateRandomNumEverySecond implements SourceFunction<Tuple2<String, Integer>> {
        private boolean isRun = true;
        private final Random random = new Random();
        private final List<String> keyList = Arrays.asList("hadoop", "spark", "flink");

        @Override
        public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
            while (this.isRun) {
                String key = keyList.get(random.nextInt(3));
                ctx.collect(Tuple2.of(key, random.nextInt(99)));
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            this.isRun = false;
        }
    }
}
